package com.margo.project.zookeeper.config;

import java.util.concurrent.CountDownLatch;

import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

/**
 * zookeeper优化 分段锁
 * @author fangxingbang
 *
 */
@Component
public class MargoZookeeperLock implements ZookeeperLock {

	private Logger logger = LoggerFactory.getLogger(this.getClass());

	private CountDownLatch countDownLatch;

	protected static final String PATH = "/lock";

	private static final String CONNECTSTRING = "127.0.0.1:2181";

	private ZkClient zkClient = new ZkClient(CONNECTSTRING);

	@Override
	public void lock(String ownedArea) {
		// TODO Auto-generated method stub
		if(tryLock(ownedArea)) {
			logger.info(Thread.currentThread().getName() +"获取到锁");
			return;
		}
//		waitLock(ownedArea);  //产生阻塞
		try {
			Thread.sleep(500);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		lock(ownedArea);
	}

	@Override
	public boolean tryLock(String ownedArea) {
		String pathListener=PATH+"/"+ownedArea;
        try {
            zkClient.createEphemeral(pathListener);
            return true;
        } catch (Exception e) {
        	logger.info(e.getMessage());
//			e.printStackTrace();
            return false;
        }
	}

	@Override
	public void unlock(String ownedArea) {
		logger.info(Thread.currentThread().getName()+"释放锁");
		String pathListener=PATH+"/"+ownedArea;
		zkClient.delete(pathListener);
	}

	@Override
	public void waitLock(String ownedArea) {
		//分段锁
		String pathListener=PATH+"/"+ownedArea;
		
		IZkDataListener izkDataListener = new IZkDataListener() {

			public void handleDataDeleted(String path) throws Exception {
				// 唤醒被等待的线程
				if (countDownLatch != null) {
					countDownLatch.countDown();
				}
			}

			public void handleDataChange(String path, Object data) throws Exception {

			}
		};
		// 注册事件
		zkClient.subscribeDataChanges(pathListener, izkDataListener);

		// 如果节点存
		if (zkClient.exists(pathListener)) {
			countDownLatch = new CountDownLatch(1);
			try {
				// 等待，一直等到接受到事件通知
				countDownLatch.await();
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
		// 删除监听
		zkClient.unsubscribeDataChanges(pathListener, izkDataListener);

	}
}
